package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.async.AsyncTableBackup;
import com.hexinfo.dmpro.sparing.dto.TableSyncConfDTO;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Service
@Slf4j
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class TableSyncServiceImpl implements TableSyncService {

    private final TableSyncConfService tableSyncConfService;
    private final TableSyncMethodService tableSyncMethodService;
    private final HotStandbyTableService hotStandbyTableService;
    private final ScanMetadataService scanMetadataService;
    private final AsyncTableBackup asyncTableBackup;
    private final MessageCenterService messageCenterService;

    /**
     * 日志保留地址
     */
    @Value("${heatBackup.tableLogPath}")
    private String tableLogPath;
    /**
     * 异步线程超时时间
     */
    @Value("${threadMonitoring}")
    private Integer threadMonitoring;


    /**
     * 热备-表级同步
     * @return
     */
    @Override
    public Boolean tableSync(String sourceName,String rangeStartDate,String rangeEndDate){
        //① 获取基础配置
        TableSyncConfDTO tableSyncConfDTO = gainConfig(rangeStartDate,rangeEndDate,sourceName);
        //② 通过源名，获取所有表集合
        List<TableSyncConf> tableSyncConfs = tableSyncConfService.queryList(sourceName);
        //③ 创建表任务对象
        List<HotStandbyTable> hotStandbyTables = allTableTask(tableSyncConfs, sourceName, tableSyncConfDTO);
        //④ 遍历表，执行表同步
        List<Future<Boolean>> futures = new ArrayList<>();
        for (HotStandbyTable hotStandbyTable : hotStandbyTables) {
            futures.add(asyncTableBackup.tableHotStandby(hotStandbyTable,tableSyncConfDTO));
        }
        //判断所有获取是否全部完成
        Boolean aBoolean = tableSyncMethodService.futureStatus(futures, threadMonitoring);
        return aBoolean;
    }

    /**
     * 热备-表级同步（失败表重跑）
     * @return
     */
    @Override
    public Boolean failTableRetry(String sourceName,String taskDate){
        //获取源最近失败的表集合
        List<HotStandbyTable> hotStandbyTables = hotStandbyTableService.querySourceName(sourceName,taskDate);
        if (hotStandbyTables == null || hotStandbyTables.isEmpty()){
            log.warn(sourceName+"【表级热备】没有失败表，不进行重跑！！！");
            return true;
        }
        HotStandbyTable hotStandbyTable1 = hotStandbyTables.get(0);
        String rangeStartDate = hotStandbyTable1.getSyncRangeStartTime();
        String rangeEndDate = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        //① 获取基础配置
        TableSyncConfDTO tableSyncConfDTO = gainConfig(rangeStartDate,rangeEndDate,sourceName);
        //② 通过源名，获取所有表集合
        List<TableSyncConf> tableSyncConfs = tableSyncConfService.queryList(sourceName);
        //③ 失败的表数据库数据清除，重新热备
        List<HotStandbyTable> hotStandbyTables1 = tableTaskFailTable(tableSyncConfs, sourceName, tableSyncConfDTO,hotStandbyTables);
        //④ 遍历表，执行表同步
        List<Future<Boolean>> futures = new ArrayList<>();
        for (HotStandbyTable hotStandbyTable : hotStandbyTables1) {
            futures.add(asyncTableBackup.tableHotStandby(hotStandbyTable,tableSyncConfDTO));
        }
        //判断所有获取是否全部完成
        Boolean aBoolean = tableSyncMethodService.futureStatus(futures, threadMonitoring);
        return aBoolean;
    }

    /**
     * 热备-表级同步（失败表重跑）(手动重跑)
     * @return
     */
    @Override
    public Boolean failTableRetryTable(List<HotStandbyTable> hotStandbyTables){
        if (hotStandbyTables == null || hotStandbyTables.isEmpty()){
            log.warn("【表级热备】没有选择失败表，不进行重跑！！！");
            return true;
        }
        HotStandbyTable hotStandbyTable1 = hotStandbyTables.get(0);
        String sourceName = hotStandbyTable1.getSourceName();
        String rangeStartDate = hotStandbyTable1.getSyncRangeStartTime();
        String rangeEndDate = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        //① 获取基础配置
        TableSyncConfDTO tableSyncConfDTO = gainConfig(rangeStartDate,rangeEndDate,sourceName);
        //② 通过源名，获取所有表集合
        List<TableSyncConf> tableSyncConfs = tableSyncConfService.queryList(sourceName);
        //③ 失败的表数据库数据清除，重新热备
        List<HotStandbyTable> hotStandbyTables1 = tableTaskFailTable(tableSyncConfs, sourceName, tableSyncConfDTO,hotStandbyTables);
        //④ 遍历表，执行表同步
        List<Future<Boolean>> futures = new ArrayList<>();
        for (HotStandbyTable hotStandbyTable : hotStandbyTables1) {
            futures.add(asyncTableBackup.tableHotStandby(hotStandbyTable,tableSyncConfDTO));
        }
        //判断所有获取是否全部完成
        Boolean aBoolean = tableSyncMethodService.futureStatus(futures, threadMonitoring);
        return aBoolean;
    }

    /**
     * 失败的表数据库数据清除，重新热备
     */
    private List<HotStandbyTable> tableTaskFailTable(List<TableSyncConf> tableSyncConfs,String sourceName,TableSyncConfDTO tableSyncConfDTO,
                                                        List<HotStandbyTable> hotStandbyTables1) {
        String taskDate = hotStandbyTables1.get(0).getTaskDate();
        List<HotStandbyTable> hotStandbyTables = new ArrayList<>();
        // 使用Collectors.toMap将List转换为Map
        Map<String, TableSyncConf> map = tableSyncConfs.stream().collect(Collectors.toMap(TableSyncConf::getDbTbName, conf -> conf));
        for (HotStandbyTable hotStandbyTable1 : hotStandbyTables1) {
            TableSyncConf tableSyncConf = map.get(hotStandbyTable1.getDbTbName());
            HotStandbyTable hotStandbyTable = new HotStandbyTable();
            hotStandbyTable.initializeData(taskDate,sourceName,tableSyncConf.getDbName(),tableSyncConf.getTbName(),
                    tableSyncConf.getDbTbName(), ScanCommonConstants.ClusterName.DZX.value,
                    hotStandbyTable1.getLogPath(),tableSyncConfDTO.getRangeStartDate(),tableSyncConfDTO.getRangeEndDate());
            hotStandbyTable.partitionData(tableSyncConf.getSourceValue(),tableSyncConf.getSourcePartition(),tableSyncConf.getDatePartition(),tableSyncConf.getHoldDataMonth());
            hotStandbyTables.add(hotStandbyTable);
        }
        //先删除失败记录，在添加新记录
        for (HotStandbyTable hotStandbyTable : hotStandbyTables1) {
            hotStandbyTableService.removeById(hotStandbyTable);
        }
        hotStandbyTableService.saveBatch(hotStandbyTables,1000);
        return hotStandbyTables;
    }

    /**
     * 获取全部表任务对象,保存数据库
     */
    private List<HotStandbyTable> allTableTask(List<TableSyncConf> tableSyncConfs,String sourceName,TableSyncConfDTO tableSyncConfDTO) {
        Date date = new Date();
        String taskDate = DateUtil.format(date, "yyyyMMdd");
        String sourceLogPath = tableLogPath.replace("{date}",taskDate) + sourceName + "/";
        List<HotStandbyTable> hotStandbyTables = new ArrayList<>();
        for (TableSyncConf tableSyncConf : tableSyncConfs) {
            String logPath = sourceLogPath + tableSyncConf.getDbTbName() + ".txt";
            HotStandbyTable hotStandbyTable = new HotStandbyTable();
            hotStandbyTable.initializeData(taskDate,sourceName,tableSyncConf.getDbName(),tableSyncConf.getTbName(),
                    tableSyncConf.getDbTbName(), ScanCommonConstants.ClusterName.DZX.value,
                    logPath,tableSyncConfDTO.getRangeStartDate(),tableSyncConfDTO.getRangeEndDate());
            hotStandbyTable.partitionData(tableSyncConf.getSourceValue(),tableSyncConf.getSourcePartition(),tableSyncConf.getDatePartition(),tableSyncConf.getHoldDataMonth());
            hotStandbyTables.add(hotStandbyTable);
        }
        hotStandbyTableService.saveBatch(hotStandbyTables,1000);
        return hotStandbyTables;
    }

    /**
     * 获取基础配置
     */
    public TableSyncConfDTO gainConfig(String rangeStartDate,String rangeEndDate,String sourceName){
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,"金桥");
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");

        if (StrUtil.isBlank(rangeStartDate) || StrUtil.isBlank(rangeEndDate)){
            //1、获取上次热备Date
            Date lastDate = messageCenterChange(sourceName);
            //上次热备时间向前偏移1小时
            rangeStartDate = DateUtil.format(DateUtil.offsetHour(lastDate, -1), "yyyy-MM-dd HH:mm:ss");
            rangeEndDate = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        }

        String rangeAllDate = gainRangeAllDate(rangeStartDate, rangeEndDate);
        TableSyncConfDTO tableSyncConfDTO =new TableSyncConfDTO()
                .setJQHdfsPrefix(jQCluster.getHdfsPrefixPath())
                .setWGHdfsPrefix(wGCluster.getHdfsPrefixPath())
                .setRangeStartDate(rangeStartDate)
                .setRangeEndDate(rangeEndDate)
                .setRangeAllDate(rangeAllDate);
        return tableSyncConfDTO;
    }

    /**
     * 获取上次热备Data
     * @return
     */
    public Date messageCenterChange(String dataName) {
        Date date = messageCenterService.maxCreateTime(
                ScanCommonConstants.ClusterName.YSJHQ.value +dataName);
        String format = DateUtil.format(date, "yyyy-MM-dd HH:mm:ss");
        return DateUtil.parse(format);
    }

    /**
     * 获取冷备范围所有日期,包含两端
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @return          2023-11-15|2023-11-16|2023-11-17
     */
    private String gainRangeAllDate(String startTime,String endTime) {
        String startDateString = DateUtil.parse(startTime).toDateStr();
        String endDateString = DateUtil.parse(endTime).toDateStr();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
        LocalDate startDate = LocalDate.parse(startDateString, formatter);
        LocalDate endDate = LocalDate.parse(endDateString, formatter);
        String strDate= "";
        while (!startDate.isAfter(endDate)) {
            strDate += startDate.format(formatter) + "|";
            startDate = startDate.plusDays(1); // 增加一天
        }
        String substringDate = strDate.substring(0, strDate.lastIndexOf("|"));
        return substringDate;
    }

}
